package com.dk.mq;

import com.aliyuncs.CommonRequest;
import com.aliyuncs.CommonResponse;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.exceptions.ServerException;
import com.aliyuncs.http.MethodType;
import com.aliyuncs.profile.DefaultProfile;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(topic = "phone-topic",consumerGroup = "codeRocketMQ")
public class CodeConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate  redisTemplate;

    @Override
    public void onMessage(String s) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
            //        获取message
                    MessageExt messageExt=list.get(0);
//                    重试次数
                    int reconsumeTimes = messageExt.getReconsumeTimes();
            //                    消息内容
                        byte[] body = messageExt.getBody();
            //                    消息ID
                        String msgId = messageExt.getMsgId();
                        String phone = new String(body);

                    DefaultProfile profile = DefaultProfile.getProfile("cn-beijing", "LTAI4GCbdwnjCGwQj9Z3tr3N", "27ih4pc1W1tqey9X4mmCy1JUotAX7L");
                    /** use STS Token
                     DefaultProfile profile = DefaultProfile.getProfile(
                     "<your-region-id>",           // The region ID
                     "<your-access-key-id>",       // The AccessKey ID of the RAM account
                     "<your-access-key-secret>",   // The AccessKey Secret of the RAM account
                     "<your-sts-token>");          // STS Token
                     **/
                    IAcsClient client = new DefaultAcsClient(profile);

                    CommonRequest request = new CommonRequest();
                    request.setSysMethod(MethodType.POST);
                    //       调用阿里发短息的域名
                    request.setSysDomain("dysmsapi.aliyuncs.com");
                    //        版本号
                    request.setSysVersion("2017-05-25");
                    //        发短信功能
                    request.setSysAction("SendSms");
                    //        接收方手机号
                    //request.putQueryParameter("PhoneNumbers",""+phone+"");
                    request.putQueryParameter("PhoneNumbers",phone);
                    //        签名
                    request.putQueryParameter("SignName","东东二手车");
                    //        短信模板
                    request.putQueryParameter("TemplateCode","SMS_162522027");

                    String numeric = RandomStringUtils.randomNumeric(6);

                    redisTemplate.boundHashOps("phone").put(phone,numeric);
                    redisTemplate.expire("phone",30, TimeUnit.MINUTES);
                    request.putQueryParameter("TemplateParam","{\"sms_code\":\""+numeric+"\"}");
                    try {
                        CommonResponse response = client.getCommonResponse(request);
                        System.out.println(response.getData());
                    }catch (ServerException e){
                        e.printStackTrace();
                    }catch (ClientException e) {
                        e.printStackTrace();
                    }
                    System.out.println(numeric);
                    /*if(reconsumeTimes > 1){
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;*/
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    }
}
